package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/4/11
 *
 *  动态表（Table）： 流式计算中，使用一张表映射一个数据流。数据流的数据是源源不断，好比表中的数据也是动态变化。
 *  连续查询: 对动态表的查询都是连续查询，意味着只要表中变化了一条数据，此时查询会重新被触发运算。
 *
 *  要导入的依赖:
 *  https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/dev/configuration/advanced/
 *
 *  ------------------
 *
 *   动态表如何获取:
 *          ①直接使用连接器，读取外部的数据源，直接映射为表
 *          ②将一个流转换为动态表
 *
 *  --------------------------
 *      sql编程需要有 对应的编程环境
 *
 *  --------------------
 *      TableAPI
 */
public class Demo1_HelloWorld
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        //表执行的环境
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        //使用表执行的环境把流转换为Table
        Table table = tableEnvironment.fromDataStream(ds);

        /*
            自动根据流中数据的类型，创建表结构。
                流中的类型是POJO，POJO的属性名就是列名，属性的对应类型就是列类型
         */
        table.printSchema();

        //打印表中的内容
        //table.execute().print();

        /*
                查询id=s1的传感器

                Expression： 代表一个运算的表达式。
                    通过 Expressions中的方法构造表达式。

                    Expressions.$("列名")： 获取某个列

         */
        table.where($("id").isEqual("s1"))
             .select($("id"),$("vc"),$("ts"))
             .execute()
             .print();



        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }


    }
}
